package com.navi.loaddata.loadService.impl;

import com.navi.loaddata.core.errorLog.ErrorCode;
import com.navi.loaddata.core.etlException.ETLException2;
import com.navi.loaddata.core.mysqlEntity.*;
import com.navi.loaddata.core.repository.impl.GreenPlumRepositoryImpl;
import com.navi.loaddata.core.repository.impl.MySqlRepositoryImpl;
import com.navi.loaddata.core.util.JdbcUtil;
import com.navi.loaddata.core.util.LoggerUtil;
import com.navi.loaddata.core.util.ProcessUtil;
import com.navi.loaddata.core.util.SendMail;
import com.navi.loaddata.loadService.IEtlConfServices2;
import org.apache.commons.dbutils.QueryRunner;
import org.apache.commons.dbutils.handlers.BeanListHandler;
import org.hibernate.Session;
import org.quartz.JobDataMap;
import org.quartz.JobDetail;
import org.quartz.JobExecutionContext;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.StringUtils;

import java.sql.Timestamp;
import java.text.SimpleDateFormat;
import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.Calendar;
import java.util.List;

/**
 * Created by Jay on 2018/3/10.
 */
@Service
@Transactional(transactionManager = "primaryPlatformTransactionManager")
public class EtlConfServices2 implements IEtlConfServices2 {

    SimpleDateFormat df = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");

    private static final Logger logger = LoggerFactory.getLogger(EtlConfServices2.class);
    @Autowired
    private MySqlRepositoryImpl dao;

    @Autowired
    private GreenPlumRepositoryImpl gpDao;


    @Override
    public List<EtlConfD> getEtlConf(String projectName, String jobGroupName, String jobName) {
        StringBuffer hql = new StringBuffer();
        hql.append("FROM EtlConfD WHERE 1=1 ");
        if (!StringUtils.isEmpty(projectName)) hql.append(" and projectName ='").append(projectName).append("'");
        if (!StringUtils.isEmpty(jobGroupName)) hql.append(" and jobGroupName ='").append(jobGroupName).append("'");
        if (!StringUtils.isEmpty(jobName)) hql.append(" and jobName ='").append(jobName).append("'");
        return dao.find(EtlConfD.class, hql.toString());
    }

    //重新获取当前JOB信息
    @Override
    public EtlConfD getNewConf(JobExecutionContext context) {
        try {
            JobDetail jobDetail = context.getJobDetail();
            JobDataMap jobDataMap = jobDetail.getJobDataMap();
            EtlConfD etlConfD = (EtlConfD) jobDataMap.get("etlConf");
            etlConfD = getNewConf(etlConfD);
            if (null == etlConfD) return null;
            if (!"N".equals(etlConfD.getRunFlg())) {
                logger.info("JOB IS RUNNING!");
                return null;
            }
            return etlConfD;
        } catch (Exception e) {
            logger.error("EtlConfD 获取JOB 信息 错误!\n" + LoggerUtil.stackTraceToString(e));
            return null;
        }
    }

    public EtlConfD getNewConf(EtlConfD etlConfD) {
        StringBuffer hql = new StringBuffer();
        try {
            hql.append("FROM EtlConfD WHERE validFlg = 'Y' ")
                    .append(" and projectName ='").append(etlConfD.getProjectName()).append("'")
                    .append(" and jobGroupName ='").append(etlConfD.getJobGroupName()).append("'")
                    .append(" and jobName ='").append(etlConfD.getJobName()).append("'")
                    .append(" and jobNo ='").append(etlConfD.getJobNo()).append("'");
            List<EtlConfD> list = dao.list(EtlConfD.class, hql.toString());
            if (list.size() > 0) return list.get(0);
            logger.error("projectName={} jobGroupName={} jobName={} jonNo={}无法找到该JOB!", etlConfD.getProjectName(), etlConfD.getJobGroupName(), etlConfD.getJobName(), etlConfD.getJobNo());
        } catch (Exception e) {
            logger.error("EtlConfD 获取JOB 信息 错误!\n" + LoggerUtil.stackTraceToString(e));
        }
        return null;
    }

    public EtlConfD getNewConf(String etl) {
        StringBuffer hql = new StringBuffer();
        String err="";
        if("opeInfo".equals(etl)) {
            hql.append("FROM EtlConfD WHERE 1=1 ")
                    .append(" and projectName ='WIP_INFO_ETL_PROD'")
                    .append(" and jobGroupName ='WIP_INFO_ETL'")
                    .append(" and jobName ='LoadShtWipTmpData'")
                    .append(" and jobNo ='0'");
            err="projectName=WIP_INFO_ETL_PROD jobGroupName=WIP_INFO_ETL jobName=LoadShtWipTmpData jonNo=0 无法找到该JOB!";
        }else if("ashtopeHis".equals(etl)){
            hql.append("FROM EtlConfD WHERE 1=1 ")
                    .append(" and projectName ='OPE_HIS_ETL_PROD'")
                    .append(" and jobGroupName ='OPE_HIS_ETL'")
                    .append(" and jobName ='LoadHisDataToGP'")
                    .append(" and jobNo ='1'");
            err="projectName=OPE_HIS_ETL_PROD jobGroupName=OPE_HIS_ETL jobName=LoadHisDataToGP jonNo=1 无法找到该JOB!";
        }else if("fshtopeHis".equals(etl)){
            hql.append("FROM EtlConfD WHERE 1=1 ")
                    .append(" and projectName ='OPE_HIS_ETL_PROD'")
                    .append(" and jobGroupName ='OPE_HIS_ETL'")
                    .append(" and jobName ='LoadHisDataToGP'")
                    .append(" and jobNo ='2'");
            err="projectName=OPE_HIS_ETL_PROD jobGroupName=OPE_HIS_ETL jobName=LoadHisDataToGP jonNo=2 无法找到该JOB!";
        }else if("cshtopeHis".equals(etl)){
            hql.append("FROM EtlConfD WHERE 1=1 ")
                    .append(" and projectName ='OPE_HIS_ETL_PROD'")
                    .append(" and jobGroupName ='OPE_HIS_ETL'")
                    .append(" and jobName ='LoadHisDataToGP'")
                    .append(" and jobNo ='3'");
            err="projectName=OPE_HIS_ETL_PROD jobGroupName=OPE_HIS_ETL jobName=LoadHisDataToGP jonNo=3 无法找到该JOB!";
        }else {
            return null;
        }
        try {
            List<EtlConfD> list = dao.list(EtlConfD.class, hql.toString());
            if (list.size() > 0) return list.get(0);
            logger.error(err);
        } catch (Exception e) {
            logger.error("EtlConfD 获取JOB 信息 错误!\n" + LoggerUtil.stackTraceToString(e));
        }
        return null;
    }

    public EtlConfD getNewConfModel(EtlConfD etlConfD) {
        StringBuffer hql = new StringBuffer();
        try {
            hql.append("FROM EtlConfD WHERE 1=1 ")
                    .append(" and projectName ='").append(etlConfD.getProjectName()).append("'")
                    .append(" and jobGroupName ='").append(etlConfD.getJobGroupName()).append("'")
                    .append(" and jobName ='").append(etlConfD.getJobName()).append("'")
                    .append(" and jobNo ='").append(etlConfD.getJobNo()).append("'");
            List<EtlConfD> list = dao.list(EtlConfD.class, hql.toString());
            if (list.size() > 0) return list.get(0);
            logger.error("projectName={} jobGroupName={} jobName={} jonNo={}无法找到该JOB!", etlConfD.getProjectName(), etlConfD.getJobGroupName(), etlConfD.getJobName(), etlConfD.getJobNo());
        } catch (Exception e) {
            logger.error("EtlConfD 获取JOB 信息 错误!\n" + LoggerUtil.stackTraceToString(e));
        }
        return null;
    }

    //获取JOB OPE LIST 信息
    public List<String> getOpesList(EtlConfD etlConfD) {
        List<String> opes = new ArrayList<>();

        if (!StringUtils.isEmpty(etlConfD.getOpe())) {
            opes.add(etlConfD.getOpe());
        } else if (!StringUtils.isEmpty(etlConfD.getOpes())) {
            String[] opeIds = etlConfD.getOpes().split(",");
            for (String ope : opeIds) {
                opes.add(ope);
            }
        }
        return opes;
    }


    //获取JOB EQPT LIST 信息
    public List<String> getEqptsList(EtlConfD etlConfD) {
        List<String> eqpts = new ArrayList<>();

        if (!StringUtils.isEmpty(etlConfD.getEqpt())) {
            eqpts.add(etlConfD.getEqpt());
        } else if (!StringUtils.isEmpty(etlConfD.getEqpts())) {
            String[] eqptIds = etlConfD.getEqpts().split(",");
            for (String eqpt : eqptIds) {
                eqpts.add(eqpt);
            }
        }
        return eqpts;
    }

    //Begin
    @Override
    public EtlConfD beginJob(EtlConfD etlConfD) {
        if (StringUtils.isEmpty(etlConfD.getQueryMinute())) return etlConfD;//QueryMinute is null
        if (!StringUtils.isEmpty(etlConfD.getRunEndTimestamp())) {//endTime is not null
            etlConfD.setRunStartTimestamp(etlConfD.getRunEndTimestamp());
            etlConfD.setRunEndTimestamp(null);
        }
        Timestamp etlTime = Timestamp.valueOf(LocalDateTime.now());
        Timestamp etlStart = etlConfD.getRunStartTimestamp();
        Calendar cal = Calendar.getInstance();
        cal.setTime(etlStart);
        cal.add(Calendar.MINUTE, etlConfD.getQueryMinute());
        Timestamp etlEnd = new Timestamp(cal.getTime().getTime());//end_time
        Calendar cal2 = Calendar.getInstance();//endTime>now+buffer
        cal2.setTime(etlTime);
        cal2.add(Calendar.MINUTE, -etlConfD.getBufferMinute());

        //实时跑
        Timestamp etlEnd2 = new Timestamp(cal2.getTime().getTime());//now-buffer
        if (etlEnd.after(etlEnd2)) {
            etlConfD.setQueryMinute(null);
            return etlConfD;
        }
//        if(!StringUtils.isEmpty(etlConfD.getFinalEndTimestamp())){//finalEnd is not null
//            if(etlConfD.getFinalEndTimestamp().before(etlEnd)) {
//                etlConfD.setQueryMinute(null);
//                return etlConfD;
//            }
//        }


        //if queryMinute<1
        if (!StringUtils.isEmpty(etlConfD.getFinalEndTimestamp()) && etlEnd.after(etlConfD.getFinalEndTimestamp())) {
            etlEnd = etlConfD.getFinalEndTimestamp();
        }


        etlConfD.setEtlTimestamp(etlTime);
        etlConfD.setRunFlg("Y");
        etlConfD.setTimeOutMinute(null);
//        System.out.println(etlConfD.getTimeOutMinute());
        dao.update(etlConfD);

        etlConfD.setRunEndTimestamp(etlEnd);
        return etlConfD;
    }

    //修改结束时间
    @Override
    public EtlConfD endJob(EtlConfD etlConfD) {
        Session session=dao.getEntityManagerSession();
        //重新获取Vaild_Flg
        EtlConfD etlConfD_old = getNewConf(etlConfD);
        if(null==etlConfD_old) return null;

        etlConfD.setValidFlg(etlConfD_old.getValidFlg());
        etlConfD.setRunFlg("N");
        if (StringUtils.isEmpty(etlConfD_old.getFinalEndTimestamp())) {
            etlConfD.setFinalEndTimestamp(etlConfD_old.getFinalEndTimestamp());
        }

        etlConfD.setBufferMinute(etlConfD_old.getBufferMinute());
        etlConfD.setEvtUser(ProcessUtil.getStringS(etlConfD_old.getEvtUser()));
        etlConfD.setEvtTime(ProcessUtil.getTimestamp(etlConfD_old.getEvtTime()));
        etlConfD.setOpeMsg(ProcessUtil.getStringS(etlConfD_old.getOpeMsg()));
        etlConfD.setReloadJobname(etlConfD_old.getReloadJobname());
        etlConfD.setReloadJobno(etlConfD_old.getReloadJobno());
        etlConfD.setEtlEndTimestamp(Timestamp.valueOf(LocalDateTime.now()));
        etlConfD.setExt1(etlConfD_old.getExt1());
        etlConfD.setExt2(etlConfD_old.getExt2());
        etlConfD.setExt3(etlConfD_old.getExt3());
        //跑到截止时间，关闭JOB
        if (!StringUtils.isEmpty(etlConfD.getFinalEndTimestamp()) && etlConfD.getRunEndTimestamp().equals(etlConfD.getFinalEndTimestamp())) {
            etlConfD.setValidFlg("N");
        }
        session.evict(etlConfD_old);
        dao.update(etlConfD);
        return etlConfD;
    }


    //add His表
    @Override
    public void addHis(EtlConfD etlConfD,String[] resMsg) {
        EtlHisD etlHis = new EtlHisD();
        etlHis.setProjectName(etlConfD.getProjectName());
        etlHis.setJobGroupName(etlConfD.getJobGroupName());
        etlHis.setJobName(etlConfD.getJobName());
        etlHis.setJobNo(etlConfD.getJobNo());
        etlHis.setValidFlg(etlConfD.getValidFlg());
        etlHis.setRunFlg(etlConfD.getRunFlg());
        etlHis.setRunStartTimestamp(etlConfD.getRunStartTimestamp());
        etlHis.setRunEndTimestamp(etlConfD.getRunEndTimestamp());
        etlHis.setFinalEndTimestamp(etlConfD.getFinalEndTimestamp());
        etlHis.setEtlTimestamp(etlConfD.getEtlTimestamp());
        etlHis.setBufferMinute(etlConfD.getBufferMinute());
        etlHis.setQueryMinute(etlConfD.getQueryMinute());
        etlHis.setCronValue(etlConfD.getCronValue());
        etlHis.setCronExpression(etlConfD.getCronExpression());
        etlHis.setCountSize(ProcessUtil.getStr2Cnt(resMsg[0]));
        etlHis.setShop(etlConfD.getShop());
        etlHis.setOpe(etlConfD.getOpe());
        etlHis.setEqpt(etlConfD.getEqpt());
        etlHis.setEvtUser(etlConfD.getEvtUser());
        etlHis.setEvtTime(etlConfD.getEvtTime());
        etlHis.setTimeOutMinute(etlConfD.getTimeOutMinute());
        etlHis.setOpeMsg(etlConfD.getOpeMsg());
        etlHis.setClassName(etlConfD.getClassName());
        etlHis.setErrorFlg(resMsg[2]);
        etlHis.setCellType(etlConfD.getCellType());
        etlHis.setExt1(ProcessUtil.getStringS(etlConfD.getExt1()));
        etlHis.setReloadJobname(ProcessUtil.getStringS(etlConfD.getReloadJobname()));
        etlHis.setReloadJobno(ProcessUtil.getStringS(etlConfD.getReloadJobno()));
        etlHis.setReloadFlg("N");
        etlHis.setOpes(etlConfD.getOpes());
        etlHis.setErrorCnt(ProcessUtil.getStr2Cnt(resMsg[1]));
        etlHis.setErrorMsg(resMsg[2]);
        etlHis.setEqpts(etlConfD.getEqpts());
        etlHis.setReloadCnt(etlConfD.getReloadCnt());
        etlHis.setEtlEndTimestamp(etlConfD.getEtlEndTimestamp());
        etlHis.setExt1(etlConfD.getExt1());
        etlHis.setExt2(etlConfD.getExt2());
        etlHis.setExt3(etlConfD.getExt3());
        dao.save(etlHis);
    }

    @Override
    public void sendEmail(EtlConfD etlConfD, String errMsg) throws Exception {
        String hql = "from EtlEmailD where projectName='" + etlConfD.getProjectName() + "'" +
                " and jobGroupName='" + etlConfD.getJobGroupName() + "'" +
                " and jobName='" + etlConfD.getJobName() + "'" +
                " and validFlg='Y'";
        try {
            List<EtlEmailD> receiverList = dao.list(EtlEmailD.class, hql);
            for (EtlEmailD receiver : receiverList) {
                SendMail.sendMail(receiver.getReceiver(), "EDA_ETL", etlConfD.getProjectName() + "_" + etlConfD.getProjectName() + "_" + etlConfD.getJobName() + "_" + etlConfD.getJobNo() + "_{" + etlConfD.getRunStartTimestamp() + "-" + etlConfD.getRunEndTimestamp() + "}", errMsg);
            }
        } catch (Exception e) {
            logger.error(LoggerUtil.stackTraceToString(e));
        }

    }

    public EtlConfD getReloadJob(EtlConfD etlConfD) throws Exception {
        QueryRunner qr_mysql = JdbcUtil.getQueryRunner_mysql();
        String hql = "select job_Id as jobId,project_Name as projectName,job_Group_Name as jobGroupName," +
                " job_Name as jobName,job_No as jobNo,valid_Flg as validFlg,run_Flg as runFlg," +
                " run_Start_Timestamp as runStartTimestamp,run_End_Timestamp as runEndTimestamp," +
                " final_End_Timestamp as finalEndTimestamp,etl_Timestamp as etlTimestamp,buffer_Minute as bufferMinute," +
                " query_Minute as queryMinute,cron_Value as cronValue,cron_Expression as cronExpression," +
                " count_Size as countSize,shop,ope,eqpt,create_User as createUser,create_Time as createTime," +
                " evt_User as evtUser,evt_Time as evtTime,time_Out_Minute as timeOutMinute,ope_Msg as opeMsg," +
                " class_Name as className,error_Flg as errorFlg,cell_Type as cellType,ext_1 as ext1," +
                " reload_Jobname as reloadJobname,reload_Flg as reloadFlg,reload_Jobno as reloadJobno," +
                " ERROR_MSG as errorMsg," +
                " reload_cnt as reloadCnt," +
                " ext_2 as ext2," +
                " ext_3 as ext3" +
                " from Etl_His_D" +
                " where reload_Jobname='EDA_RELOAD' and reload_Jobno is not null and reload_Flg='N' and ERROR_FLG is not null" +
                " AND etl_Timestamp>='"+etlConfD.getRunStartTimestamp()+"'" +
                " and reload_cnt<3" +
                " order by job_Id asc limit 1";
        List<EtlHisD> listHis = new ArrayList<>();
        try {
            listHis = qr_mysql.query(hql, new BeanListHandler<EtlHisD>(EtlHisD.class));
        } catch (Exception e) {
            throw new ETLException2(ErrorCode.ERROR_GET_RELOAD_JOB, e);
        }
        if (!listHis.isEmpty()) {
            EtlHisD etlHisD = listHis.get(0);
            String hqlReload = "from EtlConfD where projectName='EDA_ETL_PROD' and jobGroupName='EDA_RELOAD' and jobName='RELOAD'" +
                    " and jobNo='" + etlHisD.getReloadJobno() + "' and validFlg='N' and runFlg='N'";
            List<EtlConfD> reloadEtlConfDList = dao.list(EtlConfD.class, hqlReload);
            if (reloadEtlConfDList.size() > 0) {//if runflg='Y' 则不执行
                EtlConfD reloadEtlConfD = new EtlConfD();
                reloadEtlConfD = reloadEtlConfDList.get(0);
                reloadEtlConfD.setValidFlg("Y");
                reloadEtlConfD.setRunFlg("N");
                reloadEtlConfD.setRunStartTimestamp(etlHisD.getRunStartTimestamp());
                reloadEtlConfD.setRunEndTimestamp(null);
                reloadEtlConfD.setFinalEndTimestamp(etlHisD.getRunEndTimestamp());
                reloadEtlConfD.setEtlTimestamp(null);
                reloadEtlConfD.setBufferMinute(etlHisD.getBufferMinute());
                reloadEtlConfD.setQueryMinute(etlHisD.getQueryMinute());
                reloadEtlConfD.setCronExpression(etlHisD.getCronExpression());
                reloadEtlConfD.setCountSize(0);
                reloadEtlConfD.setShop(etlHisD.getShop());
                reloadEtlConfD.setOpe(etlHisD.getOpe());
                reloadEtlConfD.setEqpt(etlHisD.getEqpt());
                reloadEtlConfD.setTimeOutMinute(null);
                reloadEtlConfD.setOpeMsg("自动重跑机制");
                reloadEtlConfD.setClassName(etlHisD.getClassName());
                reloadEtlConfD.setCellType(etlHisD.getCellType());
                reloadEtlConfD.setExt1(etlHisD.getExt1());
                reloadEtlConfD.setReloadJobname(etlHisD.getReloadJobname());
                reloadEtlConfD.setReloadJobno(etlHisD.getReloadJobno());
                reloadEtlConfD.setEqpts(etlHisD.getEqpts());
                reloadEtlConfD.setExt2(etlHisD.getExt2());
                reloadEtlConfD.setExt3(etlHisD.getExt3());

                int reloadCnt=etlHisD.getReloadCnt();
                reloadCnt++;
                reloadEtlConfD.setReloadCnt(reloadCnt);

                etlHisD.setReloadFlg("Y");
//                etlHisD.setReloadCnt(reloadCnt);
                dao.update(etlHisD);
                return reloadEtlConfD;
            } else {
                logger.info("projectName={},jobGroupName={},jobName={},jobNo={} is running!", "EDA_ETL_PROD", "EDA_RELOAD", "RELOAD", etlHisD.getReloadJobno());
            }
        }else {
            logger.info("暂无需要重跑的数据");
        }
        return null;
    }

    //Defect汇总表重跑Job信息获取
    public EtlConfD getRPTReloadJob(EtlConfD etlConfD) throws Exception {
        QueryRunner qr_mysql = JdbcUtil.getQueryRunner_mysql();
        String hql = "select job_Id as jobId,project_Name as projectName,job_Group_Name as jobGroupName," +
                " job_Name as jobName,job_No as jobNo,valid_Flg as validFlg,run_Flg as runFlg," +
                " run_Start_Timestamp as runStartTimestamp,run_End_Timestamp as runEndTimestamp," +
                " final_End_Timestamp as finalEndTimestamp,etl_Timestamp as etlTimestamp,buffer_Minute as bufferMinute," +
                " query_Minute as queryMinute,cron_Value as cronValue,cron_Expression as cronExpression," +
                " count_Size as countSize,shop,ope,eqpt,create_User as createUser,create_Time as createTime," +
                " evt_User as evtUser,evt_Time as evtTime,time_Out_Minute as timeOutMinute,ope_Msg as opeMsg," +
                " class_Name as className,error_Flg as errorFlg,cell_Type as cellType,ext_1 as ext1," +
                " reload_Jobname as reloadJobname,reload_Flg as reloadFlg,reload_Jobno as reloadJobno," +
                " reload_cnt as reloadCnt," +
                " ext_2 as ext2," +
                " ext_3 as ext3" +
                " from Etl_His_D" +
                " where reload_Jobname='RPT_RELOAD' and reload_Jobno is not null and reload_Flg='N' and ERROR_FLG is not null" +
                " AND etl_Timestamp>='2018-05-19 00:0:00'" +
                " order by job_Id asc limit 1";
        List<EtlHisD> listHis = new ArrayList<>();
        try {
            listHis = qr_mysql.query(hql, new BeanListHandler<EtlHisD>(EtlHisD.class));
        } catch (Exception e) {
            throw new ETLException2(ErrorCode.ERROR_GET_RELOAD_JOB, e);
        }

        if (listHis.size() > 0) {
            EtlHisD etlHisD = listHis.get(0);
            String hqlReload = "from EtlConfD where projectName='RPT_ETL_PROD' and jobGroupName='RPT_RELOAD' and jobName='RELOAD'" +
                    " and jobNo='" + etlHisD.getReloadJobno() + "' and validFlg='N' and runFlg='N'";
            List<EtlConfD> reloadEtlConfDList = dao.list(EtlConfD.class, hqlReload);
            if (reloadEtlConfDList.size() > 0) {//if runflg='Y' 则不执行
                EtlConfD reloadEtlConfD = new EtlConfD();
                reloadEtlConfD = reloadEtlConfDList.get(0);
                reloadEtlConfD.setValidFlg("Y");
                reloadEtlConfD.setRunFlg("N");
                reloadEtlConfD.setRunStartTimestamp(etlHisD.getRunStartTimestamp());
                reloadEtlConfD.setRunEndTimestamp(null);
                reloadEtlConfD.setFinalEndTimestamp(etlHisD.getRunEndTimestamp());
                reloadEtlConfD.setEtlTimestamp(null);
                reloadEtlConfD.setBufferMinute(etlHisD.getBufferMinute());
                reloadEtlConfD.setQueryMinute(etlHisD.getQueryMinute());
                reloadEtlConfD.setCronExpression(etlHisD.getCronExpression());
                reloadEtlConfD.setCountSize(0);
                reloadEtlConfD.setShop(etlHisD.getShop());
                reloadEtlConfD.setOpe(etlHisD.getOpe());
                reloadEtlConfD.setEqpt(etlHisD.getEqpt());
                reloadEtlConfD.setTimeOutMinute(null);
                reloadEtlConfD.setOpeMsg("自动重跑机制");
                reloadEtlConfD.setClassName(etlHisD.getClassName());
                reloadEtlConfD.setCellType(etlHisD.getCellType());
                reloadEtlConfD.setExt1(etlHisD.getExt1());
                reloadEtlConfD.setReloadJobname(etlHisD.getReloadJobname());
                reloadEtlConfD.setReloadJobno(etlHisD.getReloadJobno());
                reloadEtlConfD.setEqpts(etlHisD.getEqpts());
                reloadEtlConfD.setExt2(etlHisD.getExt2());
                reloadEtlConfD.setExt3(etlHisD.getExt3());

                etlHisD.setReloadFlg("Y");
//                int reloadCnt=etlHisD.getReloadCnt();
                int reloadCnt=0;
                etlHisD.setReloadCnt(reloadCnt+1);
                dao.update(etlHisD);
                return reloadEtlConfD;
            } else {
                logger.info("projectName={},jobGroupName={},jobName={},jobNo={} is running!", "RPT_ETL_PROD", "RPT_RELOAD", "RELOAD", etlHisD.getReloadJobno());
            }
        }else {
            logger.info("暂无需要重跑的数据");
        }
        return null;
    }


    public void saveReloadJob(EtlConfD etlConfD) {
        try {
            dao.update(etlConfD);
        }catch(Exception e){
         throw new ETLException2(ErrorCode.ERROR_UPDATE_RELOAD_JOB,e);
        }
    }


    //修改结束时间 etlModel
    @Override
    public EtlConfDModel endJobOfModel(EtlConfD etlConfD) {
        Session session = dao.getEntityManagerSession();
        EtlConfDModel etlConfDModel = EtlConfDMapper.INSTANCE.etlConfDToModel(etlConfD);
        //重新获取Job相关信息
        EtlConfD etlConfD_old = getNewConfModel(etlConfD);
        if (null == etlConfD_old) return null;
        EtlConfDModel etlConfDModel_old = EtlConfDMapper.INSTANCE.etlConfDToModel(etlConfD_old);
        session.evict(etlConfD_old);

        etlConfDModel.setValidFlg(etlConfDModel_old.getValidFlg());
        etlConfDModel.setRunFlg("N");
        if (StringUtils.isEmpty(etlConfDModel_old.getFinalEndTimestamp())) {
            etlConfDModel.setFinalEndTimestamp(etlConfDModel_old.getFinalEndTimestamp());
        }
        etlConfDModel.setBufferMinute(etlConfDModel_old.getBufferMinute());
        etlConfDModel.setEvtUser(ProcessUtil.getStringS(etlConfDModel_old.getEvtUser()));
        etlConfDModel.setEvtTime(ProcessUtil.getTimestamp(etlConfDModel_old.getEvtTime()));
        etlConfDModel.setOpeMsg(ProcessUtil.getStringS(etlConfDModel_old.getOpeMsg()));
        etlConfDModel.setReloadJobname(etlConfDModel_old.getReloadJobname());
        etlConfDModel.setReloadJobno(etlConfDModel_old.getReloadJobno());
        etlConfDModel.setEtlEndTimestamp(Timestamp.valueOf(LocalDateTime.now()));
        etlConfDModel.setExt1(etlConfDModel_old.getExt1());
        etlConfDModel.setExt2(etlConfDModel_old.getExt2());
        etlConfDModel.setExt3(etlConfDModel_old.getExt3());
        return etlConfDModel;
    }

    //add His表
    @Override
    public void addHisOfModel(EtlConfDModel etlConfD, String[] resMsg) {
        int okCnt = ProcessUtil.getStr2Cnt(resMsg[0]);
        int errCnt = ProcessUtil.getStr2Cnt(resMsg[1]);
        String errMsg = resMsg[2];


        EtlHisD etlHis = new EtlHisD();
        etlHis.setProjectName(etlConfD.getProjectName());
        etlHis.setJobGroupName(etlConfD.getJobGroupName());
        etlHis.setJobName(etlConfD.getJobName());
        etlHis.setJobNo(etlConfD.getJobNo());
        etlHis.setValidFlg(etlConfD.getValidFlg());
        etlHis.setRunFlg(etlConfD.getRunFlg());
        etlHis.setRunStartTimestamp(etlConfD.getRunStartTimestamp());
        etlHis.setRunEndTimestamp(etlConfD.getRunEndTimestamp());
        etlHis.setFinalEndTimestamp(etlConfD.getFinalEndTimestamp());
        etlHis.setEtlTimestamp(etlConfD.getEtlTimestamp());
        etlHis.setBufferMinute(etlConfD.getBufferMinute());
        etlHis.setQueryMinute(etlConfD.getQueryMinute());
        etlHis.setCronValue(etlConfD.getCronValue());
        etlHis.setCronExpression(etlConfD.getCronExpression());
        etlHis.setCountSize(okCnt);
        etlHis.setShop(etlConfD.getShop());
        etlHis.setOpe(etlConfD.getOpe());
        etlHis.setEqpt(etlConfD.getEqpt());
        etlHis.setEvtUser(etlConfD.getEvtUser());
        etlHis.setEvtTime(etlConfD.getEvtTime());
        etlHis.setTimeOutMinute(etlConfD.getTimeOutMinute());
        etlHis.setOpeMsg(etlConfD.getOpeMsg());
        etlHis.setClassName(etlConfD.getClassName());
        etlHis.setErrorFlg(errMsg);
        etlHis.setCellType(etlConfD.getCellType());
        etlHis.setExt1(ProcessUtil.getStringS(etlConfD.getExt1()));
        etlHis.setReloadJobname(ProcessUtil.getStringS(etlConfD.getReloadJobname()));
        etlHis.setReloadJobno(ProcessUtil.getStringS(etlConfD.getReloadJobno()));
        if(!"RELOAD".equals(etlConfD.getJobName())) {
            etlHis.setReloadFlg(specialErr(etlConfD.getJobName(), errMsg));//非重跑机制下的JOB 连接问题，只记录不重跑
        }else{
            etlHis.setReloadFlg("N");
        }
        etlHis.setOpes(etlConfD.getOpes());
        etlHis.setErrorCnt(errCnt);
        etlHis.setErrorMsg(errMsg);
        etlHis.setEqpts(etlConfD.getEqpts());
        etlHis.setReloadCnt(etlConfD.getReloadCnt());
        etlHis.setEtlEndTimestamp(etlConfD.getEtlEndTimestamp());
        etlHis.setExt2(etlConfD.getExt2());
        etlHis.setExt3(etlConfD.getExt3());
        dao.save(etlHis);
    }


    public void resetJob(EtlConfD etlConfD) {
        etlConfD.setRunEndTimestamp(null);
        etlConfD.setValidFlg("N");
        saveReloadJob(etlConfD);

        try {
            Thread thread = Thread.currentThread();
            thread.sleep(60000);//暂停60秒后程序继续执行
        } catch (InterruptedException ex) {
            logger.error(LoggerUtil.stackTraceToString(ex));
        }

        etlConfD.setValidFlg("Y");
        etlConfD.setRunFlg("N");
        saveReloadJob(etlConfD);
    }

    public String specialErr(String jobName, String errMsg) {
        if (!StringUtils.isEmpty(errMsg) && (errMsg.contains("Unable to acquire JDBC Connection") || errMsg.contains("ORA-01683") || errMsg.contains("ORA-01688") || errMsg.contains("connect timed out")||errMsg.contains("Unexpected internal error"))) {
            return "Y";
        }

        return "N";
    }
}
